package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.bean.ProvinceStats;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/18 13:59
 */
public class DwsProvinceApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsProvinceApp().init(4003, 1, "DwsProvinceApp");
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
 
        // 1. 先消费数据: 建立动态表与Topic: dwm_order_wide
        tEnv.executeSql("create table order_wide(" +
                            "   province_id bigint, " +
                            "   province_name string, " +
                            "   province_area_code string, " +
                            "   province_iso_code string, " +
                            "   province_3166_2_code string, " +
                            "   split_total_amount decimal(20, 2), " +
                            "   order_id bigint, " +
                            "   create_time string, " +
                            "   et as to_timestamp(create_time, 'yyyy-MM-dd HH:mm:ss'), " +
                            "   watermark for et as et - interval '3' second" +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            "   'properties.group.id' = 'DwsProvinceApp', " +
                            "   'topic' = '" + Constant.TOPIC_DWM_ORDER_WIDE + "', " +
                            "   'scan.startup.mode' = 'latest-offset', " +
                            "   'format' = 'json'" +
                            ")");
        // 2. 开窗聚合
        Table result = tEnv.sqlQuery("select" +
                                        " date_format(tumble_start(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss') stt, " +
                                        " date_format(tumble_end(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss') edt, " +
                                        " province_id, " +
                                        " province_name, " +
                                        " province_area_code area_code, " +
                                        " province_iso_code iso_code, " +
                                        " province_3166_2_code iso_3166_2," +
                                        " sum(split_total_amount) order_amount, " +
                                        " count(distinct(order_id)) order_count," +
                                        " unix_timestamp()*1000 as ts " +
                                    "from order_wide " +
                                        "group by " +
                                        " tumble(et, interval '5' second), " +
                                        " province_id, " +
                                        " province_name, " +
                                        " province_area_code, " +
                                        " province_iso_code, " +
                                        " province_3166_2_code ");
    
        tEnv
            .toRetractStream(result, ProvinceStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("gmall2022", "province_stats_2022", ProvinceStats.class));
        
    }
    
           /*
           stt DateTime,
           edt DateTime,
           province_id  UInt64,
           province_name String,
           area_code String ,
           iso_code String,
           iso_3166_2 String ,
           order_amount Decimal64(2),
           order_count UInt64,
           ts UInt64
         */
}
